package jp.aist.gtrc.plus.scheduler.algorithm;

import java.security.Security;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import jp.aist.gtrc.plus.scheduler.job.JobSorter;
import jp.aist.gtrc.plus.scheduler.job.JobSortKey;
import jp.aist.gtrc.plus.scheduler.job.WaitingTooLongJobChecker;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocateRequest;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeInfo;
import jp.aist.gtrc.plus.scheduler.node.ProvisionedNodeChecker;
import jp.aist.gtrc.plus.scheduler.node.UnusedNodeChecker;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSJobStatus;
import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.util.Filter;

import provisioner.ComputeNode;
import provisioner.ComputeProvider;
import provisioner.EucaProvider;
import provisioner.NoMoreResourceException;
import provisioner.ProvisionEvent;
import provisioner.ProvisionEventListener;
import provisioner.ResourceStatusDownEvent;
import provisioner.ResourceStatusUpEvent;
import provisioner.Resource;
import provisioner.ResourceProvisioner;
import provisioner.ResourceProvisionerImpl;
import provisioner.ResourceNotMatchedException;

import org.apache.log4j.Logger;

/*
 * Job scheduling: Fifo
 * Provisioning: Most (number of ) wanted resource first 
 * Release: any resource when no job needs
 */
public class MostWantedResourceFirst implements Scheduler, ProvisionEventListener {

	protected static final Logger logger = Logger.getLogger(MostWantedResourceFirst.class);

        private JobSorter jobSorter;
        private NodeAllocator nodeAllocator;

        private ResourceProvisioner provisioner;
        private IdleNodeTracker idleNodeTracker;
        private ResourceSetMap resourceBeingProvSetMap;  // mapping from type to resources during starting and customizing process
        private ResourceSetMap resourceAlreadyProvSetMap;  // mapping from type to resources that are ready to be used

        public MostWantedResourceFirst(NodeAllocator nodeAllocator) {
           assert(nodeAllocator != null);
           logger.debug("MostWantedResouceFirst constructor()");
           this.jobSorter = new JobSorter();
           jobSorter.addJobSortKey(JobSortKey.SubmitTime);
           this.nodeAllocator = nodeAllocator;
           this.idleNodeTracker = new IdleNodeTracker();
           this.resourceBeingProvSetMap = new ResourceSetMap();
           this.resourceAlreadyProvSetMap = new ResourceSetMap();
 
           try {
              Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());
              ResourceProvisionerImpl rp = new ResourceProvisionerImpl();
              ComputeProvider computeProvider = new EucaProvider("emi-1C8F159F","/home/ekasit/src/typica-1.7/mykey.private","/home/ekasit/src/typica-1.7/ca.key","/home/ekasit/src/typica-1.7/ca.crt","m1.small","eki-8DF216F7","eri-B4CF14B6");
              rp.setComputeProvider(computeProvider);
              this.provisioner = rp;
           } catch (Exception e) {
              e.printStackTrace();
              System.exit(1);
           }
        }

        public void preSchedule(Collection<JobStatus> jobs, Collection<NodeInfo> nodes) {

        }

        public Collection<Allocation> doScheduling(Collection<JobStatus> jobs, Collection<NodeInfo> nodes) {
           List<JobStatus> sortedJobs = null;
           if (jobs != null) {    
              sortedJobs = new ArrayList<JobStatus>(jobSorter.sort(jobs));
           } else {
              sortedJobs = new ArrayList<JobStatus>();
           }
           /*
            * Trying to find a schedulable job until job list exhaust 
            * (not efficient).
            */
           Collection<Allocation> allocs = new ArrayList<Allocation>();
           while (sortedJobs.size() > 0) {
              JobStatus job = sortedJobs.get(0);
              //System.out.println("doSchedule() tests job " + job.getJobID());
              Collection<NodeInfo> allocatedNodes = nodeAllocator.suitable(nodes, new NodeAllocateRequest(job));
              if ( (allocatedNodes != null) && (allocatedNodes.size() > 0) ) {
                 allocs.add(new Allocation(job,allocatedNodes)); 
                 System.out.print("Job " + job.getJobID() + " is to be dispatched now on ");
                 for (NodeInfo node : allocatedNodes) {
                    System.out.print(node.getName() + " ");
                 }
                 System.out.println();

                 break;  // just only one job at a time
              }
              sortedJobs.remove(job);
           }

           return allocs;
        }

        public void postSchedule(Collection<JobStatus> jobs, Collection<NodeInfo> nodes, Collection<Allocation> allocs) {

           // release no longer used resources
           Filter<NodeInfo> nodeFilter = new Filter<NodeInfo>();
           nodeFilter.addChecker(new UnusedNodeChecker());
           nodeFilter.addChecker(new ProvisionedNodeChecker());
           Collection<NodeInfo> unusedNodes = nodeFilter.filter(nodes);
           Collection<NodeInfo> expiredNodes = idleNodeTracker.filter(unusedNodes);
           // exclude just allocated nodes           
           for (Allocation alloc : allocs) {
              expiredNodes.removeAll(alloc.getNodes());
           }
           
           releaseUnusedResources(provisioner,expiredNodes);

           /*
            * Check if more resources must be recruited by dynamic provisioning.
            * Happened only when no job and node is scheduled. 
            */
           if ( (jobs != null) && (jobs.size() > 0) ) {
              if ( (allocs == null) || (allocs.size() == 0) ) {
                 // recruit more resources
                 Filter<JobStatus> jobFilter = new Filter<JobStatus>();
                 jobFilter.addChecker(new WaitingTooLongJobChecker());
                 Collection<JobStatus> waitingJobs = jobFilter.filter(jobs);
                 recruitResourcesByMostWantedFirst(provisioner,waitingJobs);
              }
           }
           
           showStats(jobs);

        }

        public void actionPerformed(ProvisionEvent e) throws Exception {
           if (e.getEventType() == ProvisionEvent.RESOURCE_STATUS_UP) {
              ResourceStatusUpEvent ce = (ResourceStatusUpEvent) e;
              Collection<Resource> resources = ce.getChangedResources();
              for (Resource resource : resources) {
                 System.out.println("Got up event from " + resource.getId());
                 String resourceType = resourceBeingProvSetMap.getType(resource);
                 //
                 // should make these two an atomic
                 //
                 resourceBeingProvSetMap.removeResource(resource);
                 resourceAlreadyProvSetMap.addResource(resourceType,resource);
              }
           } else if (e.getEventType() == ProvisionEvent.RESOURCE_STATUS_DOWN) {
              ResourceStatusDownEvent ce = (ResourceStatusDownEvent) e;
              Collection<Resource> resources = ce.getChangedResources();
              for (Resource resource : resources) {
                 System.out.println("Got down event from " + resource.getId());
                 resourceBeingProvSetMap.removeResource(resource);
                 resourceAlreadyProvSetMap.removeResource(resource);
              }
           } 
        }


        protected void releaseUnusedResources(ResourceProvisioner provisioner, Collection<NodeInfo> expiredNodes) {
           if (expiredNodes == null) return;
           for (NodeInfo node : expiredNodes) {
              String resourceId = node.getName();
              try {
                 System.out.println("releasing resource " + resourceId);
System.out.println("We should disable job submission to this node");
                 provisioner.releaseResource(resourceId); // use instance name here
              } catch (Exception e) {
                 System.err.println(e);
                 break;  // stop releasing if failed
              }
           }
        }

        /*
         * may request several instances at once
         */
        private void recruitResourcesByMostWantedFirst(ResourceProvisioner provisioner, Collection<JobStatus> jobs) {
           if (jobs == null) return;

           // finding total resource requirements
           Map<String,Integer> resourceTypeReqs = new HashMap<String,Integer>();
           for (JobStatus job : jobs) {
              String other = ((PBSJobStatus) job).getOtherResourceRequests();
              if (other == null) {
                 System.err.println("Skipping job with resource type empty");
                 other = "";
                 continue;
              }
              Integer amount = resourceTypeReqs.get(other);
              if (amount == null) {
                 resourceTypeReqs.put(other,new Integer(1));
              } else {
                 resourceTypeReqs.put(other,amount+1);
              } 
           }

           // sort resource type by amount of needs
           SortedSet<String> sortedResourceType = new TreeSet<String>(new FrequencyComparator(resourceTypeReqs,FrequencyComparator.MOST_FIRST));
           sortedResourceType.addAll(resourceTypeReqs.keySet());

           // recruit new resources
           for (String resourceType : sortedResourceType) {
              if (resourceType.equals("") ) {
                 System.err.println("Skipping recruit resource type empty");
                 continue;
              }

              int amountReq = resourceTypeReqs.get(resourceType);
              // minus what is being provisioned
              int amountBeingProv = 0;
              Set<Resource> provisionedSet = resourceBeingProvSetMap.getResourcesByType(resourceType);
              if (provisionedSet != null) {
                 amountBeingProv = provisionedSet.size();
              } 
              int moreProv = amountReq-amountBeingProv;
              
              try {
                 for (int i=0; i < moreProv; i++) {
                    Resource resource = provisioner.requestResource(resourceType); // assume "other" contains resource properties
                    System.out.println("Recruiting resource " + resource.getId() + " type " + resourceType);
                    provisioner.registerListener(resource,this);

                    // add node to resource set
                    resourceBeingProvSetMap.addResource(resourceType,resource);
                 }
                 
              } catch (ResourceNotMatchedException e) {
   
                 // how to inform user if we skip this resource type.  
                 System.err.println("Resource " + resourceType + " not matched");
                 continue;

              } catch (NoMoreResourceException e) {
                 System.err.println(e); 
                 break;

              } catch (Exception e) {
                 e.printStackTrace();
                 break;   // stop all provisioning if failed
              }
           }

        }

        private void showStats(Collection<JobStatus> jobs) {

           long t = System.currentTimeMillis()/1000;
           if (jobs == null) {
              System.out.println(t + " jobNum 0" );
           } else {
              System.out.println(t + " jobNum " + jobs.size());
           }
//           for (String type : resourceBeingProvSetMap.getTypes()) {
//              for (Resource resource : resourceBeingProvSetMap.getResourcesByType(type))  {    
//                 System.out.println(t + " " + resource.getId() + ":" + type + " being provisioned");
//              }
//           }
//           for (String type : resourceAlreadyProvSetMap.getTypes()) {
//              for (Resource resource : resourceAlreadyProvSetMap.getResourcesByType(type))  {    
//                 System.out.println(t + " " + resource.getId() + ":" + type + " already provisioned");
//              }
//           }
           for (String type : resourceBeingProvSetMap.getTypes()) {
              int i = resourceBeingProvSetMap.getResourcesByType(type).size();     
              System.out.println(t + " Resource " + type + " being provisioned: " + i);
           }
           for (String type : resourceAlreadyProvSetMap.getTypes()) {
              int i = resourceAlreadyProvSetMap.getResourcesByType(type).size();     
              System.out.println(t + " Resource " + type + " already provisioned: " + i);
           }

        }

}
